package com.atguigu.flink.chapter11;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;

import static org.apache.flink.table.api.Expressions.$;

public class Flink04_Table_BaseUse_File {
       public static void main(String[] args) throws Exception {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
               env.setParallelism(1);

           // 1. 创建表的执行环境
           StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

           Schema schema = new Schema()
                   .field("id", DataTypes.STRING())
                   .field("ts", DataTypes.BIGINT())
                   .field("vc", DataTypes.INT());

           // 直接从文件读取数据，使用tableAPI

           // 创建了一个临时的动态表
            tableEnv
                    .connect(new FileSystem().path("input/sensor.txt"))
                    //后面的列与列的分割符和行与行的分割符，默认是','  "\n"
                    .withFormat(new Csv().fieldDelimiter(',').lineDelimiter("\n"))
                    .withSchema(schema)
                    .createTemporaryTable("sensor");// 临时表表名

           Table sensor = tableEnv.from("sensor");

           // 过滤sensor1
           Table result = sensor
                   .where($("id").isEqual("sensor_1"))
                   .select($("id"), $("ts"), $("vc"));
                // select 中的列名顺序  输入和输入的表必须一致，才能写入

           //创建一个动态表与文件关联，向动态表写数据，会自动写入到文件中
           tableEnv.connect(new FileSystem().path("input/out"))
                   .withFormat(new Csv())
                   .withSchema(schema)
                   .createTemporaryTable("out1");

           // 讲result表的数据插入临时表out1中
           result.executeInsert("out1");


       }
}
